//! # Redis 适配器测试
//!
//! 为 Redis 消息总线适配器提供完整的测试覆盖

use rustcloud_bus::{
    redis_adapter::{RedisMessageBus, RedisAdapterConfig},
    bus::MessageBus,
    message::Message,
    error::MessageBusError,
};
use serde_json::json;
use std::time::Duration;
use tokio::time::sleep;

/// 测试辅助工具
struct RedisTestHelper;

impl RedisTestHelper {
    /// 创建测试用的 Redis 配置
    fn create_test_config() -> RedisAdapterConfig {
        RedisAdapterConfig {
            redis_url: std::env::var("REDIS_URL")
                .unwrap_or_else(|_| "redis://localhost:6379".to_string()),
            pool_size: 2,
            connection_timeout: 3,
            command_timeout: 3,
            max_retries: 2,
            retry_interval_ms: 500,
            use_streams: true,
            consumer_group: Some("test_group".to_string()),
            consumer_name: Some("test_consumer".to_string()),
            stream_max_length: Some(1000),
        }
    }

    /// 创建测试用的 Redis 消息总线
    async fn create_test_bus() -> Result<RedisMessageBus, MessageBusError> {
        let config = Self::create_test_config();
        RedisMessageBus::with_config(config).await
    }

    /// 检查 Redis 是否可用
    async fn is_redis_available() -> bool {
        match Self::create_test_bus().await {
            Ok(bus) => bus.health_check().await.unwrap_or(false),
            Err(_) => false,
        }
    }

    /// 等待条件满足
    async fn wait_for_condition<F>(condition: F, timeout_ms: u64) -> bool 
    where
        F: Fn() -> bool + Send + 'static,
    {
        let start = std::time::Instant::now();
        let timeout_duration = Duration::from_millis(timeout_ms);

        while start.elapsed() < timeout_duration {
            if condition() {
                return true;
            }
            sleep(Duration::from_millis(10)).await;
        }
        false
    }
}

mod redis_bus_tests {
    use super::*;

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_redis_bus_creation() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        assert!(bus.health_check().await.unwrap());
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, 0);
        assert_eq!(stats.total_published, 0);
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_redis_bus_with_invalid_url() {
        let config = RedisAdapterConfig {
            redis_url: "redis://invalid:9999".to_string(),
            connection_timeout: 1,
            ..RedisTestHelper::create_test_config()
        };

        let result = RedisMessageBus::with_config(config).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_redis_health_check() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        // 健康的总线应该返回 true
        assert!(bus.health_check().await.unwrap());
        
        // 关闭后应该返回 false
        bus.close().await.unwrap();
        assert!(!bus.health_check().await.unwrap());
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_redis_bus_stats() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        let initial_stats = bus.stats().await.unwrap();
        assert_eq!(initial_stats.active_subscribers, 0);
        assert_eq!(initial_stats.total_published, 0);
        
        // 发布消息
        let message = Message::new("stats.test", json!({"test": true}));
        bus.publish(message).await.unwrap();
        
        let updated_stats = bus.stats().await.unwrap();
        assert_eq!(updated_stats.total_published, 1);
        assert!(updated_stats.uptime_seconds > 0);
    }
}

mod redis_publish_tests {
    use super::*;

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_simple_publish() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        let message = Message::new("simple.test", json!({
            "message": "Hello Redis",
            "timestamp": chrono::Utc::now().timestamp()
        }));
        
        let result = bus.publish(message).await;
        assert!(result.is_ok());
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, 1);
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_publish_different_message_types() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        // 测试不同类型的消息
        let messages = vec![
            Message::event("user.created", json!({"id": 1, "name": "Alice"})),
            Message::command("user.update", json!({"id": 1, "name": "Bob"})),
            Message::query("user.get", json!({"id": 1})),
            Message::notification("system.alert", json!({"level": "warning"})),
        ];
        
        for message in messages {
            let result = bus.publish(message).await;
            assert!(result.is_ok());
        }
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, 4);
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_publish_large_message() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        // 创建大消息（约100KB）
        let large_data = "x".repeat(100_000);
        let message = Message::new("large.test", json!({
            "data": large_data,
            "size": 100_000
        }));
        
        let result = bus.publish(message).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_publish_to_closed_bus() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        bus.close().await.unwrap();
        
        let message = Message::new("closed.test", json!({"test": true}));
        let result = bus.publish(message).await;
        
        assert!(result.is_err());
        match result.unwrap_err() {
            MessageBusError::BusShutdown => {},
            _ => panic!("应该返回 BusShutdown 错误"),
        }
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_concurrent_publishing() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = std::sync::Arc::new(RedisTestHelper::create_test_bus().await.unwrap());
        let message_count = 100;
        let concurrency = 10;
        
        let mut handles = Vec::new();
        
        for i in 0..concurrency {
            let bus_clone = bus.clone();
            let handle = tokio::spawn(async move {
                for j in 0..(message_count / concurrency) {
                    let message = Message::new(
                        &format!("concurrent.test.{}", i),
                        json!({
                            "thread": i,
                            "sequence": j,
                            "data": format!("message-{}-{}", i, j)
                        })
                    );
                    bus_clone.publish(message).await.unwrap();
                }
            });
            handles.push(handle);
        }
        
        // 等待所有任务完成
        for handle in handles {
            handle.await.unwrap();
        }
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, message_count as u64);
    }
}

mod redis_subscribe_tests {
    use super::*;

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_simple_subscribe() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        let subscriber = bus.subscribe("subscribe.test").await.unwrap();
        assert!(subscriber.is_active());
        assert_eq!(subscriber.pattern(), "subscribe.test");
        assert!(!subscriber.subscription_id().is_empty());
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, 1);
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_subscribe_and_unsubscribe() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        let subscriber = bus.subscribe("unsub.test").await.unwrap();
        let subscription_id = subscriber.subscription_id().to_string();
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, 1);
        
        bus.unsubscribe(&subscription_id).await.unwrap();
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, 0);
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_multiple_subscribers() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        let sub1 = bus.subscribe("multi.test.1").await.unwrap();
        let sub2 = bus.subscribe("multi.test.2").await.unwrap();
        let sub3 = bus.subscribe("multi.test.3").await.unwrap();
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, 3);
        
        // 验证每个订阅者有不同的ID
        assert_ne!(sub1.subscription_id(), sub2.subscription_id());
        assert_ne!(sub2.subscription_id(), sub3.subscription_id());
        assert_ne!(sub1.subscription_id(), sub3.subscription_id());
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_subscribe_to_closed_bus() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        bus.close().await.unwrap();
        
        let result = bus.subscribe("closed.test").await;
        assert!(result.is_err());
        match result.unwrap_err() {
            MessageBusError::BusShutdown => {},
            _ => panic!("应该返回 BusShutdown 错误"),
        }
    }
}

mod redis_pubsub_integration_tests {
    use super::*;

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_basic_publish_subscribe() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        // 创建订阅者
        let subscriber = bus.subscribe("basic.pubsub.test").await.unwrap();
        
        // 等待订阅者准备就绪
        sleep(Duration::from_millis(100)).await;
        
        // 发布消息
        let test_data = json!({
            "message": "Hello Redis PubSub",
            "timestamp": chrono::Utc::now().timestamp(),
            "number": 42
        });
        let message = Message::new("basic.pubsub.test", test_data.clone());
        bus.publish(message).await.unwrap();
        
        // 接收消息
        let received = subscriber.receive_timeout(2000).await.unwrap();
        assert!(received.is_some());
        
        let received_msg = received.unwrap();
        assert_eq!(received_msg.topic(), "basic.pubsub.test");
        assert_eq!(received_msg.payload()["message"], "Hello Redis PubSub");
        assert_eq!(received_msg.payload()["number"], 42);
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_multiple_messages() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        let subscriber = bus.subscribe("multi.message.test").await.unwrap();
        
        sleep(Duration::from_millis(100)).await;
        
        // 发布多条消息
        let message_count = 5;
        for i in 0..message_count {
            let message = Message::new("multi.message.test", json!({
                "sequence": i,
                "data": format!("message-{}", i)
            }));
            bus.publish(message).await.unwrap();
        }
        
        // 接收所有消息
        let mut received_count = 0;
        for _ in 0..message_count {
            if let Some(msg) = subscriber.receive_timeout(1000).await.unwrap() {
                assert_eq!(msg.topic(), "multi.message.test");
                received_count += 1;
            }
        }
        
        assert_eq!(received_count, message_count);
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_multiple_subscribers_same_topic() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        // 创建多个订阅者订阅同一主题
        let sub1 = bus.subscribe("shared.topic").await.unwrap();
        let sub2 = bus.subscribe("shared.topic").await.unwrap();
        let sub3 = bus.subscribe("shared.topic").await.unwrap();
        
        sleep(Duration::from_millis(100)).await;
        
        // 发布消息
        let message = Message::new("shared.topic", json!({
            "broadcast": true,
            "message": "This should reach all subscribers"
        }));
        bus.publish(message).await.unwrap();
        
        // 每个订阅者都应该收到消息
        let msg1 = sub1.receive_timeout(1000).await.unwrap();
        let msg2 = sub2.receive_timeout(1000).await.unwrap();
        let msg3 = sub3.receive_timeout(1000).await.unwrap();
        
        assert!(msg1.is_some());
        assert!(msg2.is_some());
        assert!(msg3.is_some());
        
        // 验证消息内容
        for msg in [msg1.unwrap(), msg2.unwrap(), msg3.unwrap()] {
            assert_eq!(msg.topic(), "shared.topic");
            assert_eq!(msg.payload()["broadcast"], true);
        }
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_receive_timeout() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        let subscriber = bus.subscribe("timeout.test").await.unwrap();
        
        // 不发布任何消息，测试超时
        let start = std::time::Instant::now();
        let result = subscriber.receive_timeout(500).await.unwrap();
        let elapsed = start.elapsed();
        
        assert!(result.is_none()); // 应该超时返回 None
        assert!(elapsed >= Duration::from_millis(450)); // 应该至少等待接近超时时间
        assert!(elapsed < Duration::from_millis(1000)); // 不应该等待太久
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_subscriber_stats() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        let subscriber = bus.subscribe("stats.test").await.unwrap();
        
        sleep(Duration::from_millis(100)).await;
        
        // 发布几条消息
        for i in 0..3 {
            let message = Message::new("stats.test", json!({"id": i}));
            bus.publish(message).await.unwrap();
        }
        
        // 接收消息
        for _ in 0..3 {
            subscriber.receive_timeout(1000).await.unwrap();
        }
        
        let stats = subscriber.stats().await.unwrap();
        assert_eq!(stats.pattern, "stats.test");
        assert!(stats.messages_received >= 3);
        assert!(stats.is_active);
        assert!(stats.last_activity > 0);
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_list_topics() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        // 发布到不同主题
        let topics = vec!["topic.1", "topic.2", "topic.3"];
        for topic in &topics {
            let message = Message::new(topic, json!({"test": true}));
            bus.publish(message).await.unwrap();
        }
        
        // 获取主题列表
        let listed_topics = bus.list_topics().await.unwrap();
        
        // 验证所有发布的主题都在列表中
        for topic in topics {
            assert!(listed_topics.contains(&topic.to_string()));
        }
    }
}

mod redis_error_handling_tests {
    use super::*;

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_connection_recovery() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        // 这个测试需要模拟网络中断，暂时简化
        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        assert!(bus.health_check().await.unwrap());
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_subscriber_close() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        let subscriber = bus.subscribe("close.test").await.unwrap();
        
        assert!(subscriber.is_active());
        
        // 关闭订阅者
        subscriber.close().await.unwrap();
        assert!(!subscriber.is_active());
        
        // 关闭后接收应该失败
        let result = subscriber.receive().await;
        assert!(result.is_err());
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_bus_close_cleanup() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = RedisTestHelper::create_test_bus().await.unwrap();
        
        // 创建多个订阅者
        let _sub1 = bus.subscribe("cleanup.test.1").await.unwrap();
        let _sub2 = bus.subscribe("cleanup.test.2").await.unwrap();
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, 2);
        
        // 关闭总线应该清理所有订阅者
        bus.close().await.unwrap();
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, 0);
    }
}

mod redis_performance_tests {
    use super::*;

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器，且是性能测试
    async fn test_high_throughput_publishing() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = std::sync::Arc::new(RedisTestHelper::create_test_bus().await.unwrap());
        let message_count = 1000;
        
        let start = std::time::Instant::now();
        
        // 并发发布消息
        let mut handles = Vec::new();
        let batch_size = 100;
        
        for batch in 0..(message_count / batch_size) {
            let bus_clone = bus.clone();
            let handle = tokio::spawn(async move {
                for i in 0..batch_size {
                    let message = Message::new(
                        "performance.test",
                        json!({
                            "batch": batch,
                            "sequence": i,
                            "timestamp": chrono::Utc::now().timestamp_millis()
                        })
                    );
                    bus_clone.publish(message).await.unwrap();
                }
            });
            handles.push(handle);
        }
        
        for handle in handles {
            handle.await.unwrap();
        }
        
        let elapsed = start.elapsed();
        let throughput = message_count as f64 / elapsed.as_secs_f64();
        
        println!("Redis 发布性能测试:");
        println!("  • 消息数量: {}", message_count);
        println!("  • 耗时: {:?}", elapsed);
        println!("  • 吞吐量: {:.2} msg/s", throughput);
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.total_published, message_count as u64);
        
        // 验证性能要求（根据实际环境调整）
        assert!(throughput > 100.0, "吞吐量应该 > 100 msg/s，实际: {:.2}", throughput);
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器，且是性能测试
    async fn test_concurrent_subscribers_performance() {
        if !RedisTestHelper::is_redis_available().await {
            println!("跳过测试：Redis 服务器不可用");
            return;
        }

        let bus = std::sync::Arc::new(RedisTestHelper::create_test_bus().await.unwrap());
        let subscriber_count = 50;
        
        // 创建多个订阅者
        let mut subscribers = Vec::new();
        for i in 0..subscriber_count {
            let subscriber = bus.subscribe(&format!("perf.test.{}", i % 10)).await.unwrap();
            subscribers.push(subscriber);
        }
        
        sleep(Duration::from_millis(100)).await;
        
        let stats = bus.stats().await.unwrap();
        assert_eq!(stats.active_subscribers, subscriber_count);
        
        println!("Redis 订阅者性能测试:");
        println!("  • 订阅者数量: {}", subscriber_count);
        println!("  • 创建成功: {}", stats.active_subscribers);
    }
}